package com.shockweb.service;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.shockweb.client.impl.ConfigCenterClient;
import com.shockweb.common.log.LogManager;
import com.shockweb.service.config.ServiceConfig;
import com.shockweb.utils.AbstractSyncThread;

/**
 * 微服务高阶限流
 * 
 * @author 彭明华
 * 2018年8月14日 创建
 */
public class ServiceThresholdManager extends AbstractSyncThread{
	
	/**
	 * 连接到注册服务器客户端
	 */
	private ServiceConfig config = null;
	
	public String configCenterGroup = null;
	/**
	 * 每个限流设置
	 */
	private Map<String,Long> serviceThresholds = new HashMap<>();
	
	/**
	 * 当前服务正在调用的次数
	 */
	private Map<String,Long> serviceDoings = new HashMap<>();
    private static final ReadWriteLock doingLock = new ReentrantReadWriteLock();
    
    /**
     * 限流值
     * @return
     */
	public Long getServiceThresholds(String service) {
        try {
        	doingLock.readLock().lock();
        	return serviceThresholds.get(service);
        } finally {
        	doingLock.readLock().unlock();
        }
	}
	
	/**
	 * 构造方法
	 * @param ConfigCenterUrls
	 */
	public ServiceThresholdManager(ServiceConfig config) {
		this.config = config;
		super.start();
	}
	
	/**
	 * 调用次数加1
	 * @param service
	 */
	public boolean addDoings(String service) {
        try {
        	doingLock.writeLock().lock();
        	long doings = 1;
        	if(serviceDoings.containsKey(service)) {
        		doings = serviceDoings.get(service)+1;
        	}
        	serviceDoings.put(service, doings);
        	if(serviceThresholds.containsKey(service) && doings > serviceThresholds.get(service)) {
        		return false;
        	}
        } finally {
        	doingLock.writeLock().unlock();
        }
        return true;
	}
	
	/**
	 * 调用次数减1
	 * @param service
	 */
	public void subDoings(String service) {
        try {
        	doingLock.writeLock().lock();
        	if(serviceDoings.containsKey(service) && serviceThresholds.containsKey(service)) {
        		serviceDoings.put(service, serviceDoings.get(service)-1);
        	}
        } finally {
        	doingLock.writeLock().unlock();
        }
	}

	

	public void run() {
		while(!Thread.currentThread().isInterrupted()) {
			try {
				String urls = config.getConfigCenterUrls();
				String group = config.getConfigCenterGroup();
	        	if(group==null){
	        		group = "service";
	        	}
			   	if(urls!=null){
		    		ConfigCenterClient client = null;
		    		try{
			    		client = ConfigCenterClient.getConfigCenterClient(urls);
			        	Map<String,Long> map = new HashMap<>();
			            try {
			            	doingLock.readLock().lock();
			            	map.putAll(serviceDoings);
			            } finally {
			            	doingLock.readLock().unlock();
			            }
			            for(String service:map.keySet()) {
			            	String value = client.getConfig(group, service + ".serviceThreshold");
			            	if(value!=null) {
			            		map.put(service,Long.parseLong(value));
			            	}
			            }
			            try {
			            	doingLock.writeLock().lock();
			            	serviceThresholds.clear();
			            	serviceThresholds.putAll(map);
			            } finally {
			            	doingLock.writeLock().unlock();
			            }
			    	}finally{
			    		if(client!=null){
			    			client.close();
			    		}
			    	}
			   	}
			}catch(Exception e) {
				LogManager.errorLog(this.getClass(), e);
			}
			if(config.getServiceThresholdSleepTime()>0){
				timeDelay(config.getServiceThresholdSleepTime());
			}
		}
	}



	
	/**
	 * 关闭服务器管理类
	 */
	public void close(){
		super.stop();
	}
}
